package com.genius.demo.rabbitmqconsumer.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.genius.demo.rabbitmqconsumer.config.RedisUtil;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.util.*;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private RedisUtil redisUtil;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //因为传递消息的时候用的map传递,所以将Map从Message内取出需要做些处理
            Map<String, String> map = objectMapper.readValue(message.getBody(), Map.class);
            System.out.println("ConvertMessage  " + map.toString());
            System.out.println("消费的主题消息来自：" + message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true);
            String modelKey = map.get("modelKey");
            List<Integer> components = new ArrayList<>();
            for (int i = 0; i < 1000000; i++) {
                components.add(i);
            }
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ObjectOutputStream os = new ObjectOutputStream(baos);
            os.writeObject(components);
            os.close();
            System.out.println("components length: " + baos.size());
            baos.close();
            long start = System.currentTimeMillis();
//            redisUtil.selectRedisDB(2);
            redisUtil.rpushAll(modelKey, components, 1000);
            System.out.println(System.currentTimeMillis() - start);

            Map<String, String> componentMsg = new HashMap<>();
            componentMsg.put("modelKey", modelKey);
            componentMsg.put("start", start + "");
            rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
            rabbitTemplate.setExchange("ModelExchange");
            rabbitTemplate.setRoutingKey("topic.component.route");
            rabbitTemplate.convertAndSend(componentMsg);

        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }

}
